package org.alis.rpc.handler;

import com.google.common.collect.Maps;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.handler.timeout.IdleStateEvent;
import lombok.extern.slf4j.Slf4j;
import org.alis.rpc.common.bean.RpcRequest;
import org.alis.rpc.common.bean.RpcResponse;
import org.alis.rpc.common.contant.RpcConstants;
import org.alis.rpc.common.protocol.RpcProtocol;
import org.alis.rpc.manager.ConnectManager;

import java.net.SocketAddress;
import java.util.Map;

/**
 * RpcClientHandler desc: rpc 客户端处理器
 *
 * @author luochuan
 * @date 2020/9/30
 */
@Slf4j
public class RpcClientHandler<V> extends SimpleChannelInboundHandler<RpcResponse<V>> {

    /**
     * rpc 请求结果集
     */
    private Map<String,RpcFuture<V>> pendRpcMap = Maps.newConcurrentMap();

    /**
     * 通道
     */
    private volatile Channel channel;

    /**
     * 网卡地址
     */
    private SocketAddress remotePeer;

    /**
     * 自定义的rpc 协议
     */
    private RpcProtocol protocol;


    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        super.channelActive(ctx);
        this.remotePeer = this.channel.remoteAddress();
    }

    


    @Override
    public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
        super.channelRegistered(ctx);
        this.channel = ctx.channel();
    }

    /**
     * 建立通道处理器
     * @param ctx 上下文
     * @param response 返回体
     * @throws Exception 异常
     */

    @Override
    public void channelRead0(ChannelHandlerContext ctx, RpcResponse<V> response) throws Exception {
        String requestId = response.getRequestId();
        log.debug("Receive response: " + requestId);
        RpcFuture<V> rpcFuture = pendRpcMap.get(requestId);
        if (rpcFuture != null) {
            pendRpcMap.remove(requestId);
            rpcFuture.done(response);
        } else {
            log.warn("Can not get pending response for request id: " + requestId);
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        log.error("Client caught exception: " + cause.getMessage());
        ctx.close();
    }

    public void close() {
        channel.writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE);
    }

    /**
     * 获取future的任务
     * @param request 请求体
     * @return 请求结构
     */
    public RpcFuture<V> sendRequest(RpcRequest request) {
        RpcFuture<V> rpcFuture = new RpcFuture<>(request);
        pendRpcMap.put(request.getRequestId(), rpcFuture);
        try {
            ChannelFuture channelFuture = channel.writeAndFlush(request).sync();
            if (!channelFuture.isSuccess()) {
                log.error("Send request {} error", request.getRequestId());
            }
        } catch (InterruptedException e) {
            log.error("Send request exception: " + e.getMessage());
        }

        return rpcFuture;
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            //Send ping
            sendRequest(RpcConstants.getBeatPing());
            log.debug("Client send beat-ping to " + remotePeer);
        } else {
            super.userEventTriggered(ctx, evt);
        }
    }

    public void setProtocol(RpcProtocol protocol) {
        this.protocol = protocol;
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        super.channelInactive(ctx);
        ConnectManager.getInstance().removeHandler(protocol);
    }
}
